/*
    kOTF Open Trace Format - a library for generating execution traces.
    Copyright 2012 Daniel K. O.

    kOTF is free software: you can redistribute it and/or modify
    it under the terms of the GNU Lesser General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    kOTF is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public License
    along with kOTF.  If not, see <http://www.gnu.org/licenses/>.
*/

#include <assert.h>

#include "Writer.hpp"

#include "EventBuffer.hpp"

namespace kotf {

    void
    Writer::ParallelQueue::push(EventBuffer *eb)
    {
        Lock guard(dataMutex);
        data.push(eb);
        emptyCondition.notify_one();
    }
    
    EventBuffer*
    Writer::ParallelQueue::pop()
    {
        std::unique_lock<mutex> guard(dataMutex);
        emptyCondition.wait(guard, [this]{ return !data.empty(); });
        assert(!data.empty());
        EventBuffer *eb = data.front();
        data.pop();
        return eb;
    }
    
    EventBuffer*
    Writer::ParallelQueue::try_pop()
    {
        Lock guard(dataMutex);
        if (data.empty())
            return 0;
        EventBuffer *b = data.front();
        data.pop();
        return b;
    }

    bool
    Writer::ParallelQueue::empty() const
    {
        Lock guard(dataMutex);
        return data.empty();
    }
    
    size_t
    Writer::ParallelQueue::size() const
    {
        Lock guard(dataMutex);
        return data.size();
    }


    Writer::Writer()
    {
        start();
    }
    
    Writer::~Writer()
    {}

    void
    Writer::pushPending(EventBuffer* eb)
    {
        assert(eb);
        pending.push(eb);
    }

    void
    Writer::pushTermination()
    {
        pending.push(0);
    }

    EventBuffer*
    Writer::popPending()
    {
        return pending.pop();
    }

    EventBuffer*
    Writer::getAvailable(StreamImpl *stream)
    {
        assert(stream);
        if (EventBuffer *eb = available.try_pop()) {
            eb->reset(stream);
            return eb;
        }
        return new EventBuffer(stream);
    }
    

    void
    Writer::recycle(EventBuffer* eb)
    {
        assert(eb);
        if (available.size() < 1024) // we don't want it to be too big
            available.push(eb);
        else
            delete eb;
    }


    namespace {
        void writer_thread(Writer* writer)
        {
            assert(writer);
            while (EventBuffer *eb = writer->popPending()) {
                eb->write();
                writer->recycle(eb);
            }
        }
    }


    void
    Writer::start()
    {
        worker = thread(writer_thread, this);
    }


    void
    Writer::stop()
    {
        pushTermination();
        if (worker.joinable())
            worker.join();
        assert(pending.empty());
        while (!available.empty())
            delete available.pop();
    }    

}

